package com.yomob.kylin.service;

import com.clearspring.analytics.util.Lists;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateManyModel;
import com.mongodb.client.model.UpdateOptions;
import com.yomob.kylin.client.MongoDBClient;
import com.yomob.kylin.facility.Consts;
import com.yomob.kylin.support.IpUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

import static com.mongodb.client.model.Filters.eq;
import static org.apache.spark.sql.functions.col;

/**
 * 统计用户当天活跃类型、活跃次数、时间段、地理位置信息
 */
public class ActiveService extends SimpleService{
    private static final Logger LOGGER = LoggerFactory.getLogger(ActiveService.class);
    private static final String COLLECTION_NAME = "daily_user_info";
    private MongoDBClient mongoDBClient;
    private IpUtil ipUtil;
    private static final Object[] COUNTERIDS = {"active"};
    public ActiveService(MongoDBClient mongoDBClient) throws IOException, URISyntaxException {
        this.mongoDBClient = mongoDBClient;
        ipUtil = new IpUtil();
    }
    @Override
    public void execute(Dataset<Row> input) throws Exception {
        Dataset<Row> dataset = input.filter(col(Consts.Fields.COUNTERID).isin("active")).select(
                Consts.Fields.UDID,
               // "device_lang",//系统语言,暂无
                Consts.Fields.USERID,
                Consts.Fields.APPID,
                Consts.Fields.DATE
                //Consts.Fields.HOUR,
                //Consts.Fields.COUNTRY
                //Consts.Fields.EXT$IP,
                //Consts.Fields.EXT$NET
        ).groupBy(
                //todo 同一个app，同一个user，在不同设备登录
                col(Consts.Fields.UDID),
                //col(Consts.Fields.USERID), col(Consts.Fields.APPID),//all app active just a device user
                col(Consts.Fields.DATE)
        ).count();
        List<Row> rowList = dataset.collectAsList();
        List<UpdateManyModel<Document>> models = Lists.newArrayList();
        models.addAll(rowList.stream().map(row -> this.convertActive(row)).collect(Collectors.toList()));
        mongoDBClient.saveToMongoBulkByBatch(models, COLLECTION_NAME);
    }

    @Override
    public Object[] getCounterids() {
        return COUNTERIDS ;
    }
    private UpdateManyModel<Document> convertActive(Row row) {
        List<Bson> matches = Lists.newArrayList();
        Bson eq = eq(Consts.UserInfoBased.DEVICE_ID, row.getAs(Consts.Fields.UDID));
        matches.add(eq);
        Bson filter = Filters.and(matches);
        Document update = new Document();
        //todo 更新活跃行为，目前存放注册时行为数据
        //update.append(Consts.Fields.APPID,row.getAs(Consts.Fields.APPID));
        //update.append(Consts.UserInfoBehavior.ACTIVE_TIME,row.getAs(Consts.Fields.HOUR));
        //update.append(Consts.MongoCommand.SET, new Document(Consts.UserInfoBehavior.DEVICE_LANG, ""));//暂无系统语言
        //update.append(Consts.MongoCommand.SET, new Document(Consts.UserInfoBehavior.DEVICE_NETWORK, row.getAs("net")));
        //update.append(Consts.MongoCommand.SET, new Document(Consts.UserInfoBehavior.DAY_PHASE, getDayPhase(row.getAs(Consts.Fields.HOUR))));
        //update.append(Consts.MongoCommand.SET, new Document(Consts.UserInfoBehavior.DAY_TYPE, getDayType(row.getAs(Consts.Fields.DATE))));
        //String ip = row.getAs("ip");
        //String[] cityInfo = ipUtil.getCityInfo(ip);
        //update.append(Consts.MongoCommand.SET, new Document(Consts.UserInfoBehavior.LOCALTION, cityInfo[0]));
        Document activeCount = new Document();
        //activeCount.append(Consts.Fields.DATE, row.getAs(Consts.Fields.DATE));
        activeCount.append(row.getAs(Consts.Fields.DATE), row.getLong(2));
        update.append(Consts.MongoCommand.ADD_TO_SET, new Document(Consts.UserInfoBehavior.ACTIVE_FREQ, activeCount));

        UpdateOptions updateOptions = new UpdateOptions().upsert(true);
        return new UpdateManyModel<>(filter, update, updateOptions);
    }
}
